package thu.infinity.commons;

import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.PriorityQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;

public class TaskScheduler {
	
	protected Thread schedulerThread;
	protected PriorityQueue<Task> taskQueue;	// sort by Task.expireTime
	protected ThreadPoolExecutor taskHandlerThreadPool;	
	protected Map<Task, Future<?>> taskFutures;
	protected Thread futureHandlerThread;
	
	public TaskScheduler() {
		taskQueue = new PriorityQueue<Task>();
		taskHandlerThreadPool = new ThreadPoolExecutor(Para.TASKSCHEDULER_THREADPOOL_SIZE, 
				Para.TASKSCHEDULER_THREADPOOL_MAXSIZE, Para.TASKSCHEDULER_THREADPOOL_KEEPALIVETIME, 
				Para.TASKSCHEDULER_THREADPOOL_TIMEUNIT, new LinkedBlockingQueue<Runnable>(),
				new ThreadPoolExecutor.CallerRunsPolicy());
		taskFutures = new HashMap<Task, Future<?>>();
		
		schedulerThread = new Thread(new Runnable() {			
			@Override
			public void run() {
				while (true) {
					boolean expired = true;
					long curTime = System.currentTimeMillis();
					while (expired) {
						//System.out.println("schedulerThread begin.");
						synchronized (taskQueue) {
							if (taskQueue.size() > 0 && taskQueue.peek().getExpireTime() <= curTime) {
								Task newTask = taskQueue.poll();
								Future<?> future = taskHandlerThreadPool.submit(newTask); 
								synchronized (taskFutures) {
									taskFutures.put(newTask, future);
								}
							}
							else {
								expired = false;
							}
						}
						//System.out.println("schedulerThread end.");
					}					
					
					try {
						Thread.sleep(1000);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}					
				}				
				
			}
		});		
		
		futureHandlerThread = new Thread(new Runnable() {			
			@Override
			public void run() {
				// TODO Auto-generated method stub	
				// Remove done (including cancelled) task future
				while (true) {
					synchronized (taskFutures) {
						Iterator<Entry<Task, Future<?>>> iterator = taskFutures.entrySet().iterator();
						while (iterator.hasNext()) {
							Entry<Task, Future<?>> entry = iterator.next();
							if (entry.getKey().getFutureRemovable() && entry.getValue() != null && entry.getValue().isDone()) {
								iterator.remove();
							}
						}
					}
					
					try {
						Thread.sleep(Para.FUTURE_HANDLER_PERIOD);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
			}
		});
	}	
	
	public void AddTask(Task task) {
		if (task != null) {
			synchronized (taskQueue) {			
				if (taskQueue.offer(task)) {
					System.out.println("   Add task to queue.");
					taskFutures.put(task, null);
					task.setScheduler(this);				
				}			
			}			
		}						
	}
	
	public void RemoveTask(Task task) {
		if (task != null) {
			synchronized (taskQueue) {
				if (taskQueue.remove(task)) {
					task.setScheduler(null);
				}			
			}			
		}				
	}
	
	public Future<?> getTaskFuture(Task task) {
		synchronized(taskFutures) {
			if (taskFutures.containsKey(task)) {
				Future<?> future = taskFutures.get(task);
				while (future == null) {
					try {
						Thread.sleep(Para.WAIT_FUTURE_TIME);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
					future = taskFutures.get(task);
				}
				return future;
			}
			else
				return null;
		}
	}
	
	public void waitUtilFinish(Task task) {
		synchronized(taskFutures) {
			if (taskFutures.containsKey(task)) {
				while (taskFutures.get(task) == null) {
					try {
						Thread.sleep(Para.WAIT_FUTURE_TIME);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
				try {
					taskFutures.get(task).get();
					taskFutures.remove(task);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (ExecutionException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		}
	}
	
	public void start() {
		//System.out.println("TaskScheduler starts.");
		schedulerThread.start();
		futureHandlerThread.start();
	}	

}
